package com.uatair.common.client;

import com.uatair.common.core.domain.AjaxResult;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import static com.uatair.common.core.domain.AjaxResult.error;
import static com.uatair.common.core.domain.AjaxResult.success;

/**
 * @author 14820
 */
@Component
public class MqttPushClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);

    @Autowired
    private PushCallback pushCallback;

    private static MqttClient client;

    private static MqttClient getClient() {
        return client;
    }

    private static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    /**
     * Connect to MQTT broker.
     *
     * @param host      MQTT broker host address.
     * @param clientID  Client ID.
     * @param username  Username for authentication.
     * @param password  Password for authentication.
     * @param timeout   Connection timeout.
     * @param keepalive Keep alive interval.
     */
    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = createConnectOptions(username, password, timeout, keepalive);
            setClient(client);
            client.setCallback(pushCallback);
            client.connect(options);
        } catch (MqttException e) {
            logger.error("Failed to connect to MQTT broker", e);
        }
    }

    private MqttConnectOptions createConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        return options;
    }

    /**
     * Publish a message to a topic.
     *
     * @param qos         Quality of Service level.
     * @param retained    Whether the message should be retained.
     * @param topic       Topic to publish to.
     * @param pushMessage Message payload.
     * @return AjaxResult indicating success or failure.
     */
    public AjaxResult publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = createMqttMessage(qos, retained, pushMessage);
        MqttTopic mqttTopic = getClient().getTopic(topic);
        if (mqttTopic == null) {
            logger.error("Topic does not exist: {}", topic);
            return error();
        }
        try {
            MqttDeliveryToken token = mqttTopic.publish(message);
            token.waitForCompletion();
            return success();
        } catch (MqttException e) {
            logger.error("Failed to publish message", e);
            return error();
        }
    }

    private MqttMessage createMqttMessage(int qos, boolean retained, String pushMessage) {
        MqttMessage message = new MqttMessage(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        return message;
    }

    /**
     * Subscribe to a topic.
     *
     * @param topic Topic to subscribe to.
     * @param qos   Quality of Service level.
     */
    public void subscribe(String topic, int qos) {
        logger.info("Subscribing to topic: {}", topic);
        try {
            getClient().subscribe(topic, qos);
        } catch (MqttException e) {
            logger.error("Failed to subscribe to topic: {}", topic, e);
        }
    }
}
